package cn.edu.flink.scala.tutorial.state.operatorstate

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala._

object ValueTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    clickStream
      .map(new RichMapFunction[Event, String] {
        // 每个实例都会维护一个num
        var num = 0L
        // Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.
        // lazy val numState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("num state", classOf[Long]))

        override def map(value: Event): String = {
          num = num + 1L
          num.toString
        }
      })
      .print()

    env.execute("ValueTest")
  }
}
